//package com.light.dp.flink.config;
//
//import com.alibaba.fastjson.JSONObject;
//import com.light.dp.flink.config.dto.*;
//import com.light.dp.flink.dto.DataChangeInfo;
//import com.light.dp.flink.enums.JdbcDriverEnum;
//import com.light.dp.flink.serialization.MysqlDeserialization;
//import com.light.dp.flink.sink.ClickHouseStockSink;
//import com.light.dp.flink.sink.IncrementalSinkHandler;
//import com.ververica.cdc.connectors.mysql.source.MySqlSource;
//import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
//import org.apache.commons.lang3.StringUtils;
//import org.apache.flink.api.common.eventtime.WatermarkStrategy;
//import org.apache.flink.api.common.restartstrategy.RestartStrategies;
//import org.apache.flink.configuration.Configuration;
//import org.apache.flink.configuration.RestOptions;
//import org.apache.flink.streaming.api.CheckpointingMode;
//import org.apache.flink.streaming.api.datastream.DataStream;
//import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
//import org.springframework.stereotype.Component;
//
//import javax.annotation.PostConstruct;
//import java.time.ZoneId;
//import java.util.*;
//import java.util.stream.Collectors;
//
//@Component
//public class MysqlFlinkCDCConfig {
//
//    @PostConstruct
//    public void init() throws Exception {
//
//        new Thread(() -> {
//// 设置 Java 系统属性，指定时区为 Asia/Shanghai
//            System.setProperty("user.timezone", "Asia/Shanghai");
//            // 配置 Flink WebUI
//            Configuration configuration = new Configuration();
//            configuration.setInteger(RestOptions.PORT, 8078);
//            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(configuration);
//            // 开启检查点机制
//            env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
//
//            // 设置固定延迟重启策略，重试 5 次，每次重试间隔 5 秒
//            env.setRestartStrategy(RestartStrategies.fixedDelayRestart(
//                    500, // 重试次数
//                    5000 // 重试间隔，单位为毫秒
//            ));
//
//            MysqlMapping collectMapping = getOneToManyMapping1();
//
//            // 为每个源表和其对应的目标表创建数据流和 Sink
//            for (SourceJDBCDatabaseConfig sourceConfig : collectMapping.getSourceConfigs()) {
//                // 获取对应的源表集合
//                List<String> sourceTables = sourceConfig.getTableMappings().stream().map(t -> sourceConfig.getDatabase() + "." + t.getSourceTable()).collect(Collectors.toList());
//
//                MySqlSource<String> source = createMySqlSource(sourceConfig, sourceTables);
//                for (TargetJDBCDatabaseConfig targetConfig : sourceConfig.getTargetConfigs()) {
//                    DataStream<String> dataStream = env.fromSource(source, WatermarkStrategy.noWatermarks(),
//                            "Source-" + sourceConfig.getHostname() + ":" + sourceConfig.getPort() + "/" + sourceConfig.getDatabase() +
//                                    targetConfig.getJobName());
////                    dataStream.addSink(new ClickHouseStockSink(targetConfig, sourceConfig.getTableMappings()));
////                    dataStream.addSink(new ClickHouseIncrementalSink(targetConfig, sourceConfig.getTableMappings()));
//                    dataStream.addSink(new IncrementalSinkHandler(targetConfig, sourceConfig.getTableMappings()));
////                    dataStream.addSink(new StockSinkHandler(targetConfig, sourceConfig.getTableMappings()));
//                }
//            }
//
//            // 执行作业
//            try {
//
//                env.execute("Multi Source Multi Table to Multi Target Multi Table");
//            } catch (Exception e) {
//                throw new RuntimeException(e);
//            }
//
//        }).start();
//    }
//
//    private static MySqlSource<String> createMySqlSource(JDBCDatabaseConfig config, List<String> sourceTables) {
//
//        return MySqlSource.<String>builder()
//                .hostname(config.getHostname())
//                .port(config.getPort())
//                .databaseList(config.getDatabase())
//                .tableList(StringUtils.join(sourceTables, ","))
//                .username(config.getUsername())
//                .password(config.getPassword())
//                // 添加时区配置
////                .serverTimeZone("Asia/Shanghai")
//                .serverTimeZone("UTC")
//                .deserializer(new JsonDebeziumDeserializationSchema())
//                .includeSchemaChanges(true)
//                .build();
//    }
//
//
//    /**
//     * 测试一源对多库
//     * 结果：多库其中一台宕机后恢复，缺少的数据会重新写入
//     * @return
//     */
//    private static MysqlMapping getOneToManyMapping() {
//        // 列映射信息
//        List<ColumnMapping> columnMappings = Arrays.asList(new ColumnMapping("id", "id", true),
//                new ColumnMapping("name","name"),
//                new ColumnMapping("age","age"));
//        // 表映射信息
//        List<TableMapping> tableMappings = Arrays.asList(new TableMapping("source_table", "target_table", columnMappings));
//
//
//
//        // 目标库配置
//        List<TargetJDBCDatabaseConfig> targetConfigs = Arrays.asList(
//                new TargetJDBCDatabaseConfig(JdbcDriverEnum.MYSQL,"192.168.1.72", 3307, "target_db1", "canal", "canal"),
//                new TargetJDBCDatabaseConfig(JdbcDriverEnum.MYSQL,"localhost", 3306, "target_db2", "root", "root")
//        );
//
//        // 源库配置
//        List<SourceJDBCDatabaseConfig> sourceConfigs = Arrays.asList(
//                new SourceJDBCDatabaseConfig("192.168.1.72", 3307, "source_db", "canal", "canal", targetConfigs, tableMappings)
//        );
//
//        return new MysqlMapping(sourceConfigs);
//    }
//
//    public static void main(String[] args) {
//        System.out.println(JSONObject.toJSONString(getOneToManyMapping()));
//    }
//
//    /**
//     * 测试多源对一库
//     * 结果：
//     * @return
//     */
//    private MysqlMapping getManyToOneMapping() {
//        // 列映射信息
//        List<ColumnMapping> columnMappings = Arrays.asList(new ColumnMapping("id", "id", true),
//                new ColumnMapping("name","name"),
//                new ColumnMapping("age","age"));
//        // 表映射信息
////        List<TableMapping> tableMappings = Arrays.asList(new TableMapping("target_table", "source_table", columnMappings));
//        List<TableMapping> tableMappings = Arrays.asList(new TableMapping("target_table", "target_table", null));
//
//
//
//        // 目标库配置
//        List<TargetJDBCDatabaseConfig> targetConfigs = Arrays.asList(
//                new TargetJDBCDatabaseConfig(JdbcDriverEnum.MYSQL,"192.168.1.72", 3307, "source_db", "canal", "canal")
//        );
//
//        // 源库配置
//        List<SourceJDBCDatabaseConfig> sourceConfigs = Arrays.asList(
//                new SourceJDBCDatabaseConfig("192.168.1.72", 3307, "target_db1", "canal", "canal", targetConfigs, tableMappings),
//                new SourceJDBCDatabaseConfig("localhost", 3306, "target_db2", "root", "root", targetConfigs, tableMappings)
//        );
//
//
//        return new MysqlMapping(sourceConfigs);
//    }
//
//
//    private MysqlMapping getOneToManyMapping1() {
//        // 列映射信息
////        List<ColumnMapping> columnMappings = Arrays.asList(new ColumnMapping("uniqueId", "uniqueId", true),
////                new ColumnMapping("buildId","buildId"),
////                new ColumnMapping("name","name"),
////                new ColumnMapping("memo","memo")
//////                new ColumnMapping("serialNo","serialNo"),
//////                new ColumnMapping("location","location"),
//////                new ColumnMapping("sort","sort")
////        );
//        // 表映射信息
//        List<TableMapping> tableMappings = Arrays.asList(new TableMapping("g_inhousing", "g_inhousing", null));
//
//
//
//        // 目标库配置
//        List<TargetJDBCDatabaseConfig> targetConfigs = Arrays.asList(
//                new TargetJDBCDatabaseConfig(JdbcDriverEnum.MYSQL,"192.168.1.65", 3306, "customer-service", "root", "root1234")
////                new TargetJDBCDatabaseConfig(JdbcDriverEnum.MYSQL,"192.168.1.65", 3306, "maintenance-service", "root", "root1234")
//        );
//
//        // 源库配置
//        List<SourceJDBCDatabaseConfig> sourceConfigs = Arrays.asList(
//                new SourceJDBCDatabaseConfig("192.168.1.65", 3306, "maintain-service", "root", "root1234", targetConfigs, tableMappings)
//        );
//
//        return new MysqlMapping(sourceConfigs);
//    }
//
//}
//
